package com.yl.util;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import com.taosdata.jdbc.TSDBDriver;
import com.yl.constant.CalType;
import com.yl.constant.CdcOperation;
import com.yl.constant.Const;
import com.yl.constant.ErrMsg;
import com.yl.entity.MultiDataEntity;
import com.yl.entity.cdc.MysqlCdcEntity;
import com.yl.flink.filter.TargetSelector;
import com.yl.flink.processor.*;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Dispatcher;
import okhttp3.OkHttpClient;
import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBConfig;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.OutputTag;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import redis.clients.jedis.Jedis;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

@Slf4j
public class FlinkUtil {

    /**
     * 获取广播数据格式描述符
     */
    public static MapStateDescriptor<String, List<MysqlCdcEntity>> getConfigDescriptor(String key) {
        return new MapStateDescriptor<>(
                key,
                BasicTypeInfo.STRING_TYPE_INFO,
                TypeInformation.of(new TypeHint<>() {
                }));
    }

    /**
     * 获取分流标签格式描述符
     */
    public static MapStateDescriptor<String, OutputTag<MultiDataEntity>> getTagsDescriptor() {
        return new MapStateDescriptor<>(
                Const.S_TAG_CAL,
                BasicTypeInfo.STRING_TYPE_INFO,
                TypeInformation.of(new TypeHint<>() {
                }));
    }

    /**
     * 更新广播数据格式状态
     */
    public static void updateState(BroadcastState<String, List<MysqlCdcEntity>> state, MysqlCdcEntity config) throws Exception {
        CdcOperation cdcOperation = config.getCdcOperation();
        String addKey = SUtil.fmt(config.getType().getTable(), config.getAddKey());
        if (state.contains(addKey)) {
            List<MysqlCdcEntity> updateState = new ArrayList<>();
            for (MysqlCdcEntity cdcEntity : state.get(addKey)) {
                // 如果是删除操作，就把状态中对应的元素删除
                if (cdcOperation.equals(CdcOperation.D)) {
                    if (!config.getDelKey().equals(cdcEntity.getDelKey())) {
                        updateState.add(cdcEntity);
                    }
                } else {
                    // 其他操作，就把状态中对应的元素更新下
                    if (config.getDelKey().equals(cdcEntity.getDelKey())) {
                        updateState.add(config);
                    } else {
                        updateState.add(cdcEntity);
                    }
                }
            }
            state.put(addKey, updateState);
        } else {
            if (!cdcOperation.equals(CdcOperation.D)) {
                state.put(addKey, ListUtil.toList(config));
            }
        }
    }

    /**
     * 获取流执行环境
     */
    public static StreamExecutionEnvironment getStreamEnv(ParameterTool params) {
        // 环境配置
        final Configuration configuration = new Configuration();
        configuration.setString(Const.FLINK_WEBUI_PORT_KEY, params.get(Const.FLINK_WEBUI_PORT));
        Optional.ofNullable(params.get(Const.FLINK_RECOVERY_DIR))
                .ifPresent(v -> {
                            if (StrUtil.isNotBlank(v) && !Const.S_NULL.equals(v)) {
                                if (FileUtil.exist(v)) {
                                    configuration.setString(Const.EXECUTION_SAVEPOINT_PATH, v);
                                } else {
                                    log.error("检查点恢复路径配置有误！");
                                }
                            }
                        }
                );
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        // 注册全局作业配置参数
        env.getConfig().setGlobalJobParameters(params);

        // 重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));

        // 启动检查点，设置每 60s 进行一次 Checkpoint 操作
        env.enableCheckpointing(params.getLong(Const.FLINK_CHECKPOINT_INTERVAL));

        // 配置状态后端
        String checkpointDir = params.get(Const.FLINK_CHECKPOINT_DIR);
        if (StrUtil.isNotBlank(checkpointDir) && !Const.S_NULL.equals(checkpointDir)) {
            // 设置本地保存（如果是windows默认盘为当前项目所在的盘）
            env.getCheckpointConfig().setCheckpointStorage(checkpointDir);
            // checkpoint 必须在 60s 内结束，否则被丢弃，默认是 10 分钟
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            // checkpoint 语义设置为 EXACTLY_ONCE，这是默认语义
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            // CheckPoint 的超时时间，checkpoints 必须在1分钟内完成，否则丢弃
            env.getCheckpointConfig().setCheckpointTimeout(60000);
            // 设置如果任务取消，系统该如何处理检查点数据
            // RETAIN_ON_CANCELLATION：如果取消任务的时候，没有加 --savepoint，系统会保留检查点数据
            // DELETE_ON_CANCELLATION：取消任务，自动是删除检查点（不建议使用）
            env
                    .getCheckpointConfig()
                    .setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            // 允许检查点失败 3 次
            env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
            // 同时并发数量
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        }
        return env;
    }

    /**
     * 获取水印生成策略
     *
     * @param maxOfOrder 最大乱序时间，单位秒
     */
    public static WatermarkStrategy<MultiDataEntity> getWatermarkStrategy(long maxOfOrder) {
        return WatermarkStrategy
                .<MultiDataEntity>forBoundedOutOfOrderness(Duration.ofSeconds(maxOfOrder))
                .withTimestampAssigner((multiDataEntity, timestamp) -> multiDataEntity.getPickTime() / 1000);
    }

    /**
     * 根据测流标签获取计算类型
     */
    public static String getCalTypeCode(OutputTag<MultiDataEntity> calTag) {
        return calTag.getId().split(Const.PTN_BAR_BOTTOM)[0];
    }

    /**
     * 根据测流标签获取计算引用指标
     */
    public static List<String> getCalRefers(OutputTag<MultiDataEntity> calTag) {
        return ListUtil.toList(calTag.getId().split(Const.PTN_BAR_BOTTOM)[1].split(Const.PTN_COMMA));
    }

    /**
     * 根据测流标签获取计算参数
     */
    public static List<String> getCalParams(OutputTag<MultiDataEntity> calTag) {
        return ListUtil.toList(calTag.getId().split(Const.PTN_BAR_BOTTOM)[2].split(Const.PTN_COMMA));
    }

    /**
     * 获取测流标签
     */
    public static OutputTag<MultiDataEntity> getTag(String calKey) {
        return new OutputTag<>(calKey) {
        };
    }

    /**
     * 获取测流标签
     */
    public static String getCalKey(String calKey) {
        return null;
    }

    /**
     * 获取全局作业参数配置
     */
    public static ParameterTool getParameterTool(RuntimeContext runtimeContext) {
        return (ParameterTool) runtimeContext.getExecutionConfig().getGlobalJobParameters();
    }

    /**
     * 获取redis配置
     */
    public static FlinkJedisPoolConfig getJedisPoolConfig(ParameterTool params) {
        return new FlinkJedisPoolConfig.Builder()
                .setHost(params.get(Const.REDIS_HOST))
                .setDatabase(params.getInt(Const.REDIS_DB))
                .setPassword(params.get(Const.REDIS_PASSWORD))
                .setPort(params.getInt(Const.REDIS_PORT))
                .build();
    }

    /**
     * 获取redis配置
     */
    public static Jedis getJedis(ParameterTool params) {
        Jedis jedis = new Jedis(
                params.get(Const.REDIS_HOST),
                params.getInt(Const.REDIS_PORT, 6377),
                params.getInt(Const.REDIS_TIMEOUT, 5000));
        jedis.auth(params.get(Const.REDIS_PASSWORD));
        jedis.select(params.getInt(Const.REDIS_DB, 12));
        return jedis;
    }

    /**
     * 获取influxdb配置
     */
    public static InfluxDBConfig getInfluxDBConfig(ParameterTool params) {
        return InfluxDBConfig
                .builder(
                        params.get(Const.INFLUXDB_WRITE_URL),
                        params.get(Const.INFLUXDB_USERNAME),
                        params.get(Const.INFLUXDB_PASSWORD),
                        params.get(Const.INFLUXDB_DATABASE)
                )
                .enableGzip(true)
                .createDatabase(true)
                .build();
    }

    /**
     * 获取 tdengine 连接
     */
    public static Connection getTDengineConn(String jdbcUrl) throws Exception {
        Properties connProps = new Properties();
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
        connProps.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
        return DriverManager.getConnection(jdbcUrl, connProps);
    }

    /**
     * 获取mqtt配置
     */
    public static MqttClient getMqttClient(ParameterTool params) throws MqttException {
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录
        options.setCleanSession(true);
        options.setConnectionTimeout(30);
        //设置断开后重新连接
        options.setAutomaticReconnect(true);
        options.setUserName(params.get(Const.MQTT_USERNAME));
        options.setPassword(params.get(Const.MQTT_PASSWORD).toCharArray());
        // 默认值 10，发布消息很多时会报错：MQTT(32202): 正在发布过多的消息
        options.setMaxInflight(1000);
        MqttClient mqttClient = new MqttClient(params.get(Const.MQTT_SERVER_URIS), StrUtil.uuid(), new MemoryPersistence());
        mqttClient.connect(options);
        return mqttClient;
    }

    /**
     * 获取ding配置
     */
    public static OkHttpClient getDingClient(ParameterTool params) throws MqttException {
        OkHttpClient.Builder builder = new OkHttpClient.Builder();
        builder.connectTimeout(10L, TimeUnit.SECONDS);
        builder.readTimeout(10L, TimeUnit.SECONDS);
        Dispatcher dispatcher = new Dispatcher();
        dispatcher.setMaxRequestsPerHost(200);
        dispatcher.setMaxRequests(200);
        builder.dispatcher(dispatcher);
        return builder.build();
    }

    /**
     * 获取钉钉机器人签名url
     */
    public static String getDingSignUrl(ParameterTool params) throws Exception {
        String signAlgorithm = "HmacSHA256";
        String baseUrl = params.get(Const.DING_BASE_URL);
        String token = params.get(Const.DING_TOKEN);
        String secret = params.get(Const.DING_SECRET);
        long timestamp = System.currentTimeMillis();
        String stringToSign = timestamp + "\n" + secret;
        Mac mac = Mac.getInstance(signAlgorithm);
        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), signAlgorithm));
        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
        return baseUrl + token + "&timestamp=" + timestamp + "&sign=" +
                URLEncoder.encode(new String(Base64.encodeBase64(signData)), StandardCharsets.UTF_8);
    }

    /**
     * 获取状态失效时间配置
     */
    public static StateTtlConfig getStateTtlConfig(org.apache.flink.api.common.time.Time time) throws Exception {
        return StateTtlConfig
                .newBuilder(time)
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
    }

    /**
     * @param tagStream 计算标签标记流
     * @param calTag    计算标签
     */
    public static SingleOutputStreamOperator<MultiDataEntity> calTypeQuotas(SingleOutputStreamOperator<MultiDataEntity> tagStream, OutputTag<MultiDataEntity> calTag) {
        SingleOutputStreamOperator<MultiDataEntity> res;
        // 获取测流
        DataStream<MultiDataEntity> calTagStream = tagStream.getSideOutput(calTag);
        // 计算类型
        String calTypeCode = FlinkUtil.getCalTypeCode(calTag);
        // 计算引用指标
        List<String> calRefers = FlinkUtil.getCalRefers(calTag);
        // 计算参数
        List<String> calParams = FlinkUtil.getCalParams(calTag);
        if (CalType.MEAN.getCode().equals(calTypeCode)) {
            // 计算平均值
            res = calMaxMeanMin(calTagStream, calParams, new MeanReduceFunc(calRefers));
        } else if (CalType.MAX.getCode().equals(calTypeCode)) {
            // 计算最大值
            res = calMaxMeanMin(calTagStream, calParams, new MaxReduceFunc(calRefers));
        } else if (CalType.MIN.getCode().equals(calTypeCode)) {
            // 计算最小值
            res = calMaxMeanMin(calTagStream, calParams, new MinReduceFunc(calRefers));
        } else if (CalType.ROC.getCode().equals(calTypeCode)) {
            // 计算变化速率
            res = calRoc(calTagStream, calRefers, calParams);
        } else if (CalType.TILT.getCode().equals(calTypeCode)) {
            // 计算倾斜角度
            res = calTilt(calTagStream, calRefers);
        } else {
            res = tagStream;
        }
        return res;
    }

    /**
     * 【1】计算最大值、平均值、最小值
     */
    private static SingleOutputStreamOperator<MultiDataEntity> calMaxMeanMin(DataStream<MultiDataEntity> calTagStream, List<String> calParams, ReduceFunction<MultiDataEntity> reduceFunction) {
        try {
            long windowSize = DtUtil.getTime(calParams.get(0), com.yl.constant.TimeUnit.S);
            // 开窗大小
            return calTagStream
                    .map(new ConvertStreamTypeFunc())
                    .setParallelism(1)
                    // 配置水印
                    .assignTimestampsAndWatermarks(FlinkUtil.getWatermarkStrategy(5))
                    // 根据测点分组
                    .keyBy(new TargetSelector())
                    // 配置窗口
                    .window(TumblingEventTimeWindows.of(Time.seconds(windowSize)))
                    // 增量聚合
                    .reduce(reduceFunction);
        } catch (RuntimeException e) {
            throw new RuntimeException(ErrMsg.CAL_MEAN_MAX_MIN.getMsg(), e);
        }
    }

    /**
     * 【2】计算变化速率
     * 1.变化速率 = abs(本次监测值 - timeSpan时间之前的监测值) / interval
     * 2.默认interval=timeSpan，可以通过计算参数配置指定；
     * 3.刚开始不足timeSpan时间，则把第一次的值作为timeSpan时间之前的监测值
     */
    private static SingleOutputStreamOperator<MultiDataEntity> calRoc(DataStream<MultiDataEntity> calTagStream, List<String> calRefers, List<String> calParams) {
        try {
            long timeSpan = DtUtil.getTime(calParams.get(0), com.yl.constant.TimeUnit.S);
            Integer interval = calParams.size() > 1 ? Integer.parseInt(calParams.get(1)) : null;
            com.yl.constant.TimeUnit timeUnit = com.yl.constant.TimeUnit.get(calParams.get(0));
            int referQuotaCode = Integer.parseInt(calRefers.get(0));
            return calTagStream
                    .keyBy(new TargetSelector())
                    .process(new CalRocFunc(timeSpan, interval, timeUnit, referQuotaCode));
        } catch (RuntimeException e) {
            throw new RuntimeException(ErrMsg.CAL_ROC.getMsg(), e);
        }
    }

    /**
     * 【3】计算倾斜角度
     * 1.根据监测的x和y方向的数据q1和q2计算所得;
     * 2.计算逻辑：q1和q2平方和的算术平方根;
     */
    private static SingleOutputStreamOperator<MultiDataEntity> calTilt(DataStream<MultiDataEntity> calTagStream, List<String> calRefers) {
        try {
            int xqc = Integer.parseInt(calRefers.get(0));
            int yqc = Integer.parseInt(calRefers.get(1));
            return calTagStream
                    .map(new CalTiltFunc(xqc, yqc));
        } catch (RuntimeException e) {
            throw new RuntimeException(ErrMsg.CAL_TILT.getMsg(), e);
        }
    }

}
